package com.atguigu.chapter5.sink;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.HttpHost;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;

/**
 * @ClassName: CategorySink
 * @Description:
 * @Author: kele
 * @Date: 2021/4/5 23:40
 **/
public class CategorySink {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        ArrayList<HttpHost> list = new ArrayList<>();

        DataStreamSource<WaterSensor> ds = env.fromCollection(waterSensors);

        ds.addSink(new mysink());

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static class mysink extends RichSinkFunction<WaterSensor> {

        private PreparedStatement statement;
        private Connection connection;


        /**
         * 获取连接
         * @param parameters
         *
         */
        @Override
        public void open(Configuration parameters) throws Exception {

            connection = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test", "root", "aaaaaa");

            String sql = "replace into sensor values(?,?,?)";

            statement = connection.prepareStatement(sql);


        }

        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {

            statement.setString(1,value.getId());
            statement.setLong(2,value.getTs());
            statement.setInt(3,value.getVc());

            statement.execute();
        }

        @Override
        public void close() throws Exception {

            if (statement != null) {
                statement.close();
            }

            if (connection != null) {
                connection.close();
            }
        }
    }

}
